package com.lhd.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.lhd.bean.PageAnalysisBean;
import com.lhd.utils.MyClickHouseUtil;
import com.lhd.utils.MyKafkaUtil;
import com.lhd.utils.MyKafkaUtil2;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class DwsPageAnalysisToCk {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> source = env.addSource(MyKafkaUtil2.getFlinkKafkaConsumer("dws_traffic_page_analysis", "dws_traffic_page_analysis_group"));

        // 在 DwsPageAnalysisToCk 类的 map 函数中添加空值检查


        // 解析JSON数据
        DataStream<PageAnalysisBean> pageAnalysisStream = source.map(new MapFunction<String, PageAnalysisBean>() {
            @Override
            public PageAnalysisBean map(String value) throws Exception {

                if (value == null || value.trim().isEmpty()) {
                    return new PageAnalysisBean(); // 返回默认实例而不是null
                }

                JSONObject jsonObject = JSONObject.parseObject(value);
                PageAnalysisBean pageAnalysis = new PageAnalysisBean();
                // 设置各个字段值
                pageAnalysis.setPageId(jsonObject.getString("page_id"));
                pageAnalysis.setWindowStart(jsonObject.getString("window_start"));
                pageAnalysis.setWindowEnd(jsonObject.getString("window_end"));
                pageAnalysis.setVisitorCount(jsonObject.getLongValue("visitor_count"));
                pageAnalysis.setClickUserCount(jsonObject.getLongValue("click_user_count"));
                pageAnalysis.setGuideOrderUserCount(jsonObject.getLongValue("guide_order_user_count"));
                pageAnalysis.setGuidePaymentUserCount(jsonObject.getLongValue("guide_payment_user_count"));
                pageAnalysis.setGuidePaymentAmount(jsonObject.getDoubleValue("guide_payment_amount"));

                return pageAnalysis;
            }
        });

        pageAnalysisStream.print();
        // 使用工具类写入ClickHouse
        String sql = "INSERT INTO dws_traffic_page_analysis VALUES(?,?,?,?,?,?,?,?)";
        pageAnalysisStream.addSink(MyClickHouseUtil.getSinkFunction(sql));


        env.execute("DwsPageAnalysisToCk");
    }
}
